/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.es.api.client;

import com.pingan.haofang.searchcloud.api.client.ClientFactory;
import com.pingan.haofang.searchcloud.api.matedata.IndexMeta;
import com.pingan.haofang.searchcloud.api.matedata.IndexMetadataHolder;
import com.pingan.haofang.searchcloud.api.matedata.IndexMetadataHolderFactory;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.net.InetAddress;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author huhailiang937
 * @date 2017年7月5日下午5:09:41
 * @since 1.0.0
 */
@Service("esClientFactory")
public class EsClientFactory implements ClientFactory<TransportClient> {

    protected static Logger LOG = LoggerFactory.getLogger(EsClientFactory.class);

    @Value("${es.user.authorization:}")
    private String authorization;


    private static final ConcurrentHashMap<String, TransportClient> clientMap = new ConcurrentHashMap<String, TransportClient>();

    /*
     * (non-Javadoc)
     * 
     * @see
     * com.pingan.haofang.searchcloud.api.client.ClientFactory#getByName(java.
     * lang.String)
     */
    @Override
    public TransportClient getByName(String indexName) throws RuntimeException {
        IndexMetadataHolder metadataHolder = IndexMetadataHolderFactory.get();
        IndexMeta meta = metadataHolder.findIndexMeta(indexName);
        if (null == meta) {
            return null;
        }
        return getByHost(meta.getHost());
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * com.pingan.haofang.searchcloud.api.client.ClientFactory#getByHost(java.
     * lang.String)
     */
    @Override
    public TransportClient getByHost(String host) throws RuntimeException {
        TransportClient client = clientMap.get(host);

        if (null != client) {
            return client;
        }

        // 初始化地址
        String[] addresses = host.split(",");
        InetSocketTransportAddress[] transportAddresses = new InetSocketTransportAddress[addresses.length];
        for (int i = 0; i < addresses.length; i++) {
            String[] addressItems = addresses[i].split(":");
            try {
                transportAddresses[i] = new InetSocketTransportAddress(InetAddress.getByName(addressItems[0]), Integer.valueOf(addressItems[1]));
            } catch (Exception e) {
                LOG.error(e.getMessage(), e);
                throw new EsOperationException(e);
            }
        }
//        Settings setting = Settings.builder()
//                // 不允许自动刷新地址列表
//                .put("client.transport.sniff", false)
//                .put("client.transport.ignore_cluster_name", true)
//                .build();

        Settings.Builder settingBuilder = Settings.builder()
                // 不允许自动刷新地址列表
                .put("client.transport.sniff", false)
                .put("client.transport.ignore_cluster_name", true);
        if (StringUtils.isNotBlank(authorization)) {
            settingBuilder = settingBuilder.put("xpack.security.user", authorization);
            Settings settings = settingBuilder.build();

            PreBuiltXPackTransportClient preBuiltTransportClient = new PreBuiltXPackTransportClient(settings);
            client = preBuiltTransportClient.addTransportAddresses(transportAddresses);
            clientMap.put(host, client);

            return client;
        } else {
            Settings settings = settingBuilder.build();
            PreBuiltTransportClient preBuiltTransportClient = new PreBuiltTransportClient(settings);
            client = preBuiltTransportClient.addTransportAddresses(transportAddresses);
            clientMap.put(host, client);

            return client;
        }
    }

}
